package com.example.testkafkaproto.consumer.controller;

import com.example.testkafkaproto.consumer.config.ConsumerConfiguration;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;

@RestController
@RequestMapping("/consumer")
public class ConsumerController {

    @RequestMapping("/start")
    @ResponseBody
    public Map<String,Object> start() {

        Map<String,Object> appResult = new HashMap<>();
        new ConsumerConfiguration().kafkaConsumer();
        appResult.put("status","SUCCESS");
        return appResult;
    }

    @RequestMapping("/close")
    @ResponseBody
    public Map<String,Object> close() {

        Map<String,Object> appResult = new HashMap<>();
        ConsumerConfiguration.isRunninng.set(false);
        appResult.put("status","SUCCESS");
        return appResult;
    }

    @RequestMapping("/pause")
    @ResponseBody
    public Map<String,Object> pause() {

        Map<String,Object> appResult = new HashMap<>();
        ConsumerConfiguration.isPausing.set(true);
        appResult.put("status","SUCCESS");
        return appResult;
    }

    @RequestMapping("/resume")
    @ResponseBody
    public Map<String,Object> resume() {

        Map<String,Object> appResult = new HashMap<>();
        ConsumerConfiguration.isResuming.set(true);
        appResult.put("status","SUCCESS");
        return appResult;
    }

    @RequestMapping("/seek")
    @ResponseBody
    public Map<String,Object> seek(String topic,int partition,int offset) {

        Map<String,Object> appResult = new HashMap<>();
        ConsumerConfiguration.isSeeking.set(true);
        ConsumerConfiguration.seekingTopic = topic;
        ConsumerConfiguration.seekingPartition = partition;
        ConsumerConfiguration.seekingOffset = offset;
        appResult.put("status","SUCCESS");
        return appResult;
    }
}
